// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-12-09

package managers

import (
	"context"
	"gitee.com/go-libs/log"
	"gitee.com/go-libs/runtime"
	"gitee.com/gomq/sr"
	"gomq/app"
	"gomq/app/models"
	"gomq/app/services"
	"sync"
	"time"
)

var publishingPool = &sync.Pool{
	New: func() any {
		return &publishing{}
	},
}

type (
	// Publishing
	// is an interface used to publish payloads to topics of the message
	// queue like rabbitmq, rocketmq and so on.
	Publishing interface {
		// Do process received message.
		Do()

		// Release put consuming instance to pool.
		Release()
	}

	publishing struct {
		msg *sr.Payload
		sm  ServerManager
	}
)

// NewPublishing get a publishing instance from pool.
func NewPublishing(sm ServerManager, msg *sr.Payload) Publishing {
	return publishingPool.
		Get().(*publishing).
		acquirer(sm, msg)
}

func (o *publishing) Do() {
	var (
		ctx       = log.NewContextWithHeader(o.msg.Header)
		err       error
		messageId string
		t         = time.Now()
	)

	log.Infofc(ctx, `[publishing-manager] payload popped from bucket: %s`, o.msg)
	runtime.GetCounter().MqCounter().IncrProducedCount(1)

	if messageId, err = o.DoPublish(ctx); o.sm.Cluster().Store.Enabled() {
		o.DoSave(ctx, time.Now().Sub(t), messageId, err)
	}
}

func (o *publishing) DoPublish(ctx context.Context) (messageId string, err error) {
	ctx = log.NewContextForChildInfo(ctx, `[publishing-manager] call servlet to publish: plugin="%s"`, o.sm.Cluster().Plugin)
	messageId, err = o.sm.Servlet().Producer().Publish(ctx, o.msg)
	return
}

func (o *publishing) DoSave(ctx context.Context, duration time.Duration, messageId string, reason error) {
	var (
		affects int64
		bean    *models.Payloads
		err     error
		service = services.NewPayloadsService()
	)

	ctx = log.NewContextForChildInfo(ctx, `[publishing-manager] save payload to database`)

	// Trigger when save done.
	defer func() {
		if r := recover(); r != nil {
			log.Fatalfc(ctx, `[publishing-manager] saving caused an panic: %v`, r)
		}
	}()

	// Read previous bean from database.
	if bean, err = service.GetByHash(ctx, o.msg.Hash); err != nil {
		return
	}

	// Update latest status if exists.
	if bean != nil {
		if reason != nil {
			affects, err = service.SetStatusAsFailed(ctx, bean.Id, reason)
		} else {
			affects, err = service.SetStatusAsSucceed(ctx, bean.Id, messageId)
		}
		if err == nil {
			log.Infofc(ctx, `[publishing-manager] save completed: primary-key="%d", affects="%d"`, bean.Id, affects)
		}
		return
	}

	// Creates a new result.
	req := &models.Payloads{
		DelaySeconds: o.msg.DelaySeconds,
		Duration:     duration.Milliseconds(),
		Hash:         o.msg.Hash, MessageId: messageId,
		Topic: o.msg.Topic, Tag: o.msg.Tag,
		MessageTime: o.msg.Time.Format(time.DateTime),
		MessageBody: o.msg.Body,
	}
	if reason != nil {
		req.Status = app.PayloadStatusFailed
		req.ResponseError = reason.Error()
	} else {
		req.Status = app.PayloadStatusSucceed
	}

	// Load trace id from context.
	if tracing := log.ContextTracing(ctx); tracing != nil {
		req.TraceId = tracing.GetTraceId().String()
	}

	if bean, err = service.Add(ctx, req); err == nil {
		log.Infofc(ctx, `[publishing-manager] save completed: primary-key="%d"`, bean.Id)
	}
}

func (o *publishing) Release() { o.releaser() }

// + Access

func (o *publishing) acquirer(sm ServerManager, msg *sr.Payload) *publishing {
	o.sm, o.msg = sm, msg
	return o
}

func (o *publishing) releaser() {
	o.sm, o.msg = nil, nil
	publishingPool.Put(o)
}
